package com.atguigu.day08;

import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

public class Flink02_State_OpBroadCastState {

    public static void main(String[] args) throws Exception {

        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);

        //2.读取数据创建主流,读取数据创建配置流
        DataStreamSource<String> socketTextStream = env.socketTextStream("hadoop102", 9999);
        DataStreamSource<String> channelStream = env.socketTextStream("hadoop102", 8888);

        //3.将配置流广播
        MapStateDescriptor<String, String> mapStateDescriptor = new MapStateDescriptor<>("map-state",
                String.class,
                String.class);
        BroadcastStream<String> broadcastStream = channelStream.broadcast(mapStateDescriptor);

        //4.将两个流进行连接
        BroadcastConnectedStream<String, String> connectedStream = socketTextStream.connect(broadcastStream);

        //5.根据广播状态处理数据
        SingleOutputStreamOperator<String> result = connectedStream.process(new BroadcastProcessFunction<String, String, String>() {

            //处理主流数据
            @Override
            public void processElement(String value, ReadOnlyContext ctx, Collector<String> out) throws Exception {

                //获取状态
                ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

                //获取状态中的数据
                String channel = broadcastState.get("channel");

                //根据状态中的配置信息处理数据
                if ("1".equals(channel)) {
                    out.collect("1" + ":" + value);
                } else if ("2".equals(channel)) {
                    out.collect("2" + ":" + value);
                } else {
                    out.collect("other" + ":" + value);
                }
            }

            //处理广播流数据
            @Override
            public void processBroadcastElement(String value, Context ctx, Collector<String> out) throws Exception {

                //获取广播状态
                BroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

                //将数据写入状态
                broadcastState.put("channel", value);
            }
        });

        //6.打印
        result.print();

        //7.启动
        env.execute();

    }

}
